#ifndef _H_XQUEUE_HPP_
#define _H_XQUEUE_HPP_

#pragma once

#include "XThread.hpp"

namespace XUtil {
	template <class TMsg>
	class BlockingQueue
	{
	public:
		BlockingQueue() { }
		BlockingQueue(const TMsg&) = delete;
		BlockingQueue& operator=(const TMsg&) = delete;
		~BlockingQueue() { }

		void clear()
		{
			std::lock_guard<std::mutex> lock(mutex_);
			while (!queue_.empty())
			{
				queue_.pop();
			}

		}
		void notify(bool all = true)
		{
			std::lock_guard<std::mutex> lock(mutex_);
			if (all) {
				cv_.notify_all();
			}
			else {
				cv_.notify_one();
			}
		}
		void push(const TMsg& o)
		{
			std::lock_guard<std::mutex> lock(mutex_);
			queue_.push(o);
			cv_.notify_one();
		}
		bool try_pop(TMsg& o)
		{
			if (queue_.empty()) {
				return false;
			}
			std::unique_lock<std::mutex> lock(mutex_);
			if (queue_.empty()) {
				return false;
			}
			o = queue_.front();
			queue_.pop();
			return true;
		}
		bool pop(TMsg& o, size_t timeout)
		{
			std::unique_lock<std::mutex> lock(mutex_);
			if (timeout) {
				if (!cv_.wait_for(lock, std::chrono::milliseconds(timeout),
					[this]() { return !queue_.empty(); })) {
					return false;
				}
			}
			else {
				if (queue_.empty()) {
					return false;
				}
			}
			o = queue_.front();
			queue_.pop();
			return true;
		}
		bool empty()
		{
			return queue_.empty();
		}
	private:
		std::queue<TMsg> queue_;
		std::mutex mutex_;
		std::condition_variable cv_;
	};

	template <class TMsg>
	class MultiBlockingQueue
	{
	public:
		MultiBlockingQueue(size_t multi) : queues_(multi) { }
		MultiBlockingQueue(const TMsg&) = delete;
		MultiBlockingQueue& operator=(const TMsg&) = delete;
		~MultiBlockingQueue() { }

		void clear()
		{
			std::lock_guard<std::mutex> lock(mutex_);
			while (!queue_.empty())
			{
				queue_.pop();
			}
			for (auto& queue : queues_)
			{
				while (!queue.empty())
				{
					queue.pop();
				}
			}
		}
		void notify(bool all = true)
		{
			std::lock_guard<std::mutex> lock(mutex_);
			if (all) {
				cv_.notify_all();
			}
			else {
				cv_.notify_one();
			}
		}
		void push(const TMsg& o, size_t index = (size_t)-1)
		{
			std::lock_guard<std::mutex> lock(mutex_);
			if (index < queues_.size()) {
				queues_[index].push(o);
			}
			else {
				queue_.push(o);
			}
			cv_.notify_all();
		}
		bool try_pop(TMsg& o, size_t index = (size_t)-1)
		{
			if (index < queues_.size()) {
				if (queues_[index].empty() && queue_.empty()) {
					return false;
				}
				std::unique_lock<std::mutex> lock(mutex_);
				if (!queues_[index].empty()) {
					o = queues_[index].front();
					queues_[index].pop();
					return true;
				} else if (!queue_.empty()) {
					o = queue_.front();
					queue_.pop();
					return true;
				}
				return false;
			}
			else {
				if (queue_.empty()) {
					return false;
				}
				std::unique_lock<std::mutex> lock(mutex_);
				if (queue_.empty()) {
					return false;
				}
				o = queue_.front();
				queue_.pop();
				return true;
			}
		}
		bool pop(TMsg& o, size_t timeout, size_t index = (size_t)-1)
		{
			std::unique_lock<std::mutex> lock(mutex_);
			if (index < queues_.size()) {
				if (timeout) {
					if (!cv_.wait_for(lock, std::chrono::milliseconds(timeout),
						[this, index]() { return !(queues_[index].empty() && queue_.empty()); })) {
						return false;
					}
				}
				else {
					if (queues_[index].empty() && queue_.empty()) {
						return false;
					}
				}
				if (!queues_[index].empty()) {
					o = queues_[index].front();
					queues_[index].pop();
					return true;
				}
				else if (!queue_.empty()) {
					o = queue_.front();
					queue_.pop();
					return true;
				}
				return false;
			}
			else {
				if (timeout) {
					if (!cv_.wait_for(lock, std::chrono::milliseconds(timeout),
						[this]() { return !queue_.empty(); })) {
						return false;
					}
				}
				else {
					if (queue_.empty()) {
						return false;
					}
				}
				o = queue_.front();
				queue_.pop();
				return true;
			}
		}
		bool empty(size_t index = (size_t)-1)
		{
			if (index < queues_.size()) {
				return queues_[index].empty() && queue_.empty();
			}
			else {
				return queue_.empty();
			}
		}
	private:
		std::queue<TMsg> queue_;
		std::vector<std::queue<TMsg>> queues_;
		std::mutex mutex_;
		std::condition_variable cv_;
	};

	template <class T, class TMsg>
	class MultiThreadQueueT
	{
	public:
		MultiThreadQueueT(size_t timeout = 1000) : timeout_(timeout){}

		inline bool IsRun() { return !stop_flag_; }
		inline size_t GetThreadCount() { return threads_.size(); }

		void Start(size_t threads = std::thread::hardware_concurrency() + 1)
		{
			stop_flag_ = false;
			queue_ = std::make_shared<MultiBlockingQueue<TMsg>>(threads);
			for (size_t i = 0; i < threads; ++i) {
				threads_.emplace_back(
					[this, i] {
					T* pT = static_cast<T*>(this);
					pT->Run(i);
				});
			}
		}

		void Stop()
		{
			stop_flag_ = true;
			queue_->notify();
			for (auto &thread : threads_) {
				thread.join();
			}
			threads_.clear();
			queue_->clear();
			queue_.reset();
		}

		void Run(size_t index)
		{
			T* pT = static_cast<T*>(this);
			do
			{
				TMsg msg;
				if (queue_->pop(msg, timeout_, index)) {
					pT->Handle(msg);
				}
				else if(!stop_flag_) {
					pT->Idle();
				}
			} while (!stop_flag_);
		}

		void Idle()
		{

		}

		void Handle(const TMsg& msg)
		{

		}

		void Post(const TMsg& msg, size_t index = (size_t)-1)
		{
			queue_->push(msg, index);
		}

		void Broadcast(const TMsg& msg)
		{
			for (size_t i = 0; i < threads_.size(); i++)
			{
				queue_->push(msg, i);
			}
		}

	private:
		bool stop_flag_ = true;
		std::vector<std::thread> threads_;
		std::shared_ptr<MultiBlockingQueue<TMsg>> queue_;
		size_t timeout_;
	};
}

#endif//_H_XQUEUE_HPP_